package lucene.server.common.queue;

import lombok.extern.slf4j.Slf4j;
import lucene.server.common.event.EventData;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author jinji
 */
@Slf4j
public class ChangeProcessor   {

    private static ConcurrentHashMap<String, FileQueue<EventData>> fileQueueMap = new ConcurrentHashMap<>();


    public static void process(List<EventData> dataList) {
        if (dataList==null ||dataList.isEmpty()){
            return;
        }
        String topic = "cdc_"+dataList.get(0).getDatabase()+"_"+dataList.get(0).getTable();

        if (!fileQueueMap.containsKey(topic)){
            log.info("cdc task parse has handler error , topic:{}",topic);
            return;
        }
        FileQueue<EventData> fileQueue = fileQueueMap.get(topic);
        dataList.forEach(data->{
            fileQueue.put(data);
        });
        dataList.clear();
    }

    public static FileQueue<EventData> getFileQueue(String database, String table) {
        String topic = "cdc_"+database+"_"+table;

        if (!fileQueueMap.containsKey(topic)){
            try {
                FileQueue<EventData> fileQueue = FileQueue.ordinary(EventData.class, topic);
                fileQueueMap.put(topic,fileQueue);
            }catch (Exception e){
                return null;
            }
        }
        return fileQueueMap.get(topic);
    }
}
